/**  
	 * Project Name:NetSyncConsumer  
	 * File Name:StationWebTicketMessageQueue.java  
	 * Package Name:service.impl  
	 * Date:2018年5月16日下午4:31:07  
	 * Copyright (c) 2018, chenzhou1025@126.com All Rights Reserved.  
	 *  
	*/  
	  
	package service.impl;  

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;

import kafka.javaapi.consumer.ConsumerConnector;
import play.Play;
import service.MessagesQueue;
import utils.LogUtil;
import bp.DataSynBP;
	/**  
 * ClassName:StationWebTicketMessageQueue <br/>  
 * Function: TODO ADD FUNCTION. <br/>  
 * Reason:   TODO ADD REASON. <br/>  
 * Date:     2018年5月16日 下午4:31:07 <br/>  
 * @author   swnl  
 * @version    
 * @since    JDK 1.6  
 * @see        
 */
public class StationWebTicketMessageQueue implements MessagesQueue {
	
	private static Queue<String> queue = new LinkedList<String>();
	private Integer consumethreshold=Integer.parseInt(Play.configuration.getProperty("SYNC_CONSUMER_BATCH_SIZE", "1000"));
	private static ConsumerConnector consumer;
	private static boolean canconsumer = true;
	public static void setConsumer(ConsumerConnector consumer) {
		StationWebTicketMessageQueue.consumer = consumer;
	}

	private static StationWebTicketMessageQueue stationmessagequeue;
	
	public static StationWebTicketMessageQueue getInstance(){
		synchronized(StationWebTicketMessageQueue.class){
			if(stationmessagequeue==null){
				stationmessagequeue = new StationWebTicketMessageQueue();
//				stationmessagequeue.consumethreshold = SystemParam.SYNC_CONSUMER_BATCH_SIZE;
			}
		}
		return stationmessagequeue;
	}

	@Override
	public void addMessage(String message) throws Exception {
		synchronized(queue){
			queue.offer(message);
		}
		checkqueuesize();
	}

	@Override
	public List<String> getMessages(int size) throws Exception {
		// TODO Auto-generated method stub
		return null;
	}
	
	private StationWebTicketMessageQueue(Integer consumethreshold) {
		super();
		this.consumethreshold = consumethreshold;
	}
	private StationWebTicketMessageQueue() {
		super();
	}
	public static boolean canwork(){
		return canconsumer;
	}
	private int checkqueuesize() throws Exception{
		try {
			if(queue.size()>0 && (queue.size()%consumethreshold==0)){
				consumeMessages(1);
			}
		} catch (Exception e) {
			LogUtil.logerror("consumeMessages（）消费批量消息失败", e);
		}
		return 0;
	}
	
	public  void consumeMessages(int type) throws Exception{
		//queue.clear();
		//int index = consumethreshold;
		synchronized(queue){
			try {
				canconsumer = false;
				String message;
				List<String> messages = new ArrayList<String>();
				while((message=queue.poll())!=null){
					messages.add(message);
					   //index--;
				}
				if(messages.size()>0){
					DataSynBP datasynbp = new DataSynBP();
					datasynbp.stationwebticketmessagesyn(messages);
					consumer.commitOffsets();
				}
			} catch (Exception e) {
				LogUtil.logerror("consumeMessages（）消费批量消息失败", e);
			}finally{
				canconsumer = true;
			}
		}
		
	}
}
  
	